package spark

import fun.functions.{getKafkaData, openSpark, sendKafka}

object streamKafka2 {
  def main(args: Array[String]): Unit = {
    val streamingContext = openSpark
    val lines = getKafkaData(streamingContext)

    // 更新状态
    val updateGenderCount = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.sum
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }


    val semesterCount = lines
      .map { line =>
        val semester = line.split("\t")(5).toInt
        (semester, 1)
      }
      .updateStateByKey(updateGenderCount)


    semesterCount.foreachRDD { rdd =>
      val semesterCountList = rdd.collect().toList
      val record: StringBuilder = new StringBuilder("{")
      semesterCountList.foreach {
        case (semester, count) =>
          record.append(s""""$semester":$count,""")
      }
      record.setLength(record.length - 1)
      record.append("}")
      sendKafka("result1", record)
    }

    streamingContext.start()
    streamingContext.awaitTermination()

  }
}
